package com.qf.device.service.impl;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.qf.device.Dto.CmdParamDto;
import com.qf.device.entity.DeviceTopic;
import com.qf.device.entity.MqttUser;
import com.qf.device.mapper.DeviceTopicMapper;
import com.qf.device.mapper.MqttUserMapper;
import com.qf.device.service.DeviceTopicService;
import com.qf.tenant.entity.Tenant;
import com.qf.vo.R;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.stream.Collectors;

/**
 * <p>
 * Topic主题表 服务实现类
 * </p>
 *
 * @author wxh
 * @since 2023-10-27
 */
@Service
public class DeviceTopicServiceImpl extends ServiceImpl<DeviceTopicMapper, DeviceTopic> implements DeviceTopicService {
    @Autowired
    private MqttUserMapper mqttUserMapper;

    @Override
    public R sendCmd(CmdParamDto cmdParamDto) {

        //比较deviceKey，去数据库中查询对应的数据
        QueryWrapper<DeviceTopic> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().eq(DeviceTopic::getDeviceKey,cmdParamDto.getDeviceKey());

        //stream流过滤ctrl 然后转成集合
        List<DeviceTopic> deviceTopics = this.baseMapper.selectList(queryWrapper).stream().filter(item -> item.getTopic().contains("ctrl")).collect(Collectors.toList());

        //发送消息
        //1、实例化mqtt客户端
        int qos = 2;
        String broker = "tcp://47.110.143.177:1883";
        String clientId = cmdParamDto.getDeviceKey();
        MemoryPersistence persistence = new MemoryPersistence();
        try {
            MqttClient client = new MqttClient(broker,clientId,persistence);
            Tenant tenant = (Tenant) SecurityContextHolder.getContext().getAuthentication().getPrincipal();
            QueryWrapper<MqttUser> qw = new QueryWrapper<>();
            qw.lambda().eq(MqttUser::getTenantId,tenant.getId());
            MqttUser mqttUser = mqttUserMapper.selectOne(qw);

            //2、mqtt 连接选项
            MqttConnectOptions connectOptions = new MqttConnectOptions();
            connectOptions.setUserName(mqttUser.getUsername());

            String sub = StrUtil.sub(deviceTopics.get(0).getTopic(), 27, -13);
            connectOptions.setPassword(sub.toCharArray());

            //保留会话
            connectOptions.setCleanSession(true);

            //3、建立连接
            client.connect(connectOptions);

            //4、发送数据
            String send = JSONUtil.toJsonStr(cmdParamDto);

            MqttMessage mqttMessage = new MqttMessage(send.getBytes());
            mqttMessage.setQos(qos);

            client.publish(deviceTopics.get(0).getTopic(),mqttMessage);

            return R.ok();

        } catch (MqttException e) {
            throw new RuntimeException(e);
        }

    }
}
